package org.xcorpio.java8.chapter11;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class PipeliningAysnchronousTasks {

    public static void main(String[] args) {
        List<Shop> shops = Arrays.asList(new Shop("Tmall"), new Shop("JingDong"), new Shop("JuMei"), new Shop("SuNing"),
                new Shop("Paipai"));
        String product = "Cook";
        long start = System.currentTimeMillis();
        // List<String> prices = findPricesSeq(shops, product);
        ExecutorService executor = Executors.newFixedThreadPool(Math.min(shops.size(), 100), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setDaemon(true);
                return thread;
            }
        });
        // List<String> prices = findPricesAsync(shops, product, executor);
        // System.out.println(prices);
        @SuppressWarnings("rawtypes")
        CompletableFuture[] futures = findPricesStream(shops, product, executor)
                .map(future -> future.thenAccept(
                        s -> System.out.println(s + "(takes " + (System.currentTimeMillis() - start) + " ms)")))
                .toArray(size -> new CompletableFuture[size]);
        CompletableFuture.allOf(futures).join();
        System.out.println("takes:" + (System.currentTimeMillis() - start) + "ms");
    }

    public static Stream<CompletableFuture<String>> findPricesStream(List<Shop> shops, String product,
            Executor executor) {
        return shops.stream()
                .map(shop -> CompletableFuture.supplyAsync(() -> shop.getPrice(product), executor))
                .map(future -> future.thenApply(Quote::parse))
                .map(future -> future.thenCompose(
                        quote -> CompletableFuture.supplyAsync(() -> Discount.applyDiscount(quote), executor)));
    }

    public static List<String> findPricesAsync(List<Shop> shops, String product, Executor executor) {
        List<CompletableFuture<String>> priceFutures = shops.stream()
                // Asynchronously retrieve the non-discounted price from each
                // shop.
                .map(shop -> CompletableFuture.supplyAsync(() -> shop.getPrice(product), executor))
                // Transform the String returned by a shop into a Quote object
                // when it becomes available.
                .map(future -> future.thenApply(Quote::parse))
                // Compose the resulting Future with another asynchronous task,
                // applying the discount code.
                .map(future -> future.thenCompose(
                        quote -> CompletableFuture.supplyAsync(() -> Discount.applyDiscount(quote), executor)))
                .collect(Collectors.toList());
        return priceFutures.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList());
    }

    public static List<String> findPricesSeq(List<Shop> shops, String product) {
        return shops.stream()
                .map(shop -> shop.getPrice(product))
                .peek(System.out::println)
                .map(Quote::parse)
                .peek(System.out::println)
                .map(Discount::applyDiscount)
                .peek(System.out::println)
                .collect(Collectors.toList());
    }
}
